package com.jeff.android.lab.architecture.rxjava2.api;

import android.util.Log;

import com.jeff.android.lab.BaseFragment;
import com.jeff.android.lab.R;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import butterknife.OnClick;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.internal.operators.observable.ObservableFromCallable;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.schedulers.Schedulers;

/**
 * @author: duff
 * @date: 2019/3/18
 * @since: 1.0.0
 */
public class RxJava2ApiFragment extends BaseFragment {

    @Override
    protected int getRootLayout() {
        return R.layout.rxjava2_fragment_api;
    }

    @OnClick(R.id.bt1)
    public void simpleUse() {
        Observable<String> observable = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onNext("2");
                        emitter.onComplete();
                        emitter.onNext("3");
                        Log.d(TAG, "Observable emitter end");
                    }
                }
        );

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe ...");
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG, "onNext :: " + s);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError ...");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete ...");
            }
        };

        observable.subscribe(observer);
    }

    @OnClick(R.id.bt2)
    public void disposable() {
        Observable<String> observable = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onNext("2");
                        emitter.onNext("3");
                        emitter.onComplete();
                        Log.d(TAG, "Observable emitter end");
                    }
                }
        );

        Observer<String> observer = new Observer<String>() {
            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe ...");
                disposable = d;
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG, "onNext :: " + s);

                if ("2".equals(s)) {
                    disposable.dispose();
                }
                Log.d(TAG, "isDisposed:" + disposable.isDisposed());

            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError ...");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete ...");
            }
        };

        observable.subscribe(observer);
    }

    @OnClick(R.id.bt3)
    public void chainTypeCall() {
        Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onNext("2");
                        emitter.onComplete();
                        Log.d(TAG, "Observable emitter end");
                    }
                }
        ).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe ...");
            }

            @Override
            public void onNext(String s) {
                Log.d(TAG, "onNext :: " + s);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError ...");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete ...");
            }
        });
    }


    @OnClick(R.id.bt4)
    public void consumerUse() {
        // 只接收 onNext 事件
        Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onComplete();
                        Log.d(TAG, "Observable emitter end");
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s);
            }
        });

        // 接收 onNext 和 onError 事件
        Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onError(new Exception("Exception error"));
                        Log.d(TAG, "Observable emitter end");
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s);
            }

        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "accept error :: " + throwable.getMessage());
            }
        });

        // 接收 onNext、onError、onComplete 事件
        Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onComplete();
                        Log.d(TAG, "Observable emitter end");
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s);
            }

        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "accept error :: " + throwable.getMessage());
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.d(TAG, "onComplete ... ");
            }
        });

        // 接收 onNext、onError、onComplete、disposable 事件
        Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe");
                        emitter.onNext("1");
                        emitter.onComplete();
                        Log.d(TAG, "Observable emitter end");
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s);
            }

        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(TAG, "accept error :: " + throwable.getMessage());
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.d(TAG, "onComplete ... ");
            }
        }, new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.d(TAG, "accept disposable ...  ");
            }
        });
    }


    @OnClick(R.id.bt5)
    public void threadScheduler() {
        Observable<String> observable = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable subscribe :: " + Thread.currentThread().getName());
                        emitter.onNext("Thread scheduler");
                    }
                }
        );

        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s + " :: " + Thread.currentThread().getName());
            }
        };

        Log.d(TAG, "不使用线程调度器 ...");
        observable.subscribe(consumer);

        Log.d(TAG, "使用线程调度器 ...");
        observable
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);


        Log.d(TAG, "指定多个线程调度器：");
        observable
                .subscribeOn(Schedulers.io())
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .subscribe(consumer);
        Log.d(TAG, "上游线程只有第一次指定的有效，下游线程最终会切换至最后一个指定的线程");
    }

    @OnClick(R.id.bt6)
    public void createOperator() {
        // create 创建一个被观察者 (需要主动调用onNext来发送事件)
        Log.d(TAG, "使用 create 创建 Observable");
        Observable.create(
                new ObservableOnSubscribe<String>() { // 创建 ObservableOnSubscribe 并重写其 subscribe 方法，就可以通过 ObservableEmitter 发射器向观察者发送事件
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("1");
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s);
            }
        });

        // just 创建一个被观察者，并发送事件，发送的事件不可以超过10个以上
        Log.d(TAG, "使用 just 创建 Observable");
        Observable.just(1, 3, 5)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept :: " + integer);
                    }
                });

        // fromArray 这个方法和 just() 类似，只不过 fromArray 可以传入多于10个的变量，并且可以传入一个数组
        Log.d(TAG, "使用 fromArray 创建 Observable");
        Observable.fromArray(new int[]{2, 4, 8})
                .subscribe(new Consumer<int[]>() {
                    @Override
                    public void accept(int[] ints) throws Exception {
                        for (int i : ints) {
                            Log.d(TAG, "accept :: " + i);
                        }
                    }
                });

        // fromCallable
        // 这里的 Callable 是 java.util.concurrent 中的 Callable，Callable 和 Runnable 的用法基本一致，只是它会返回一个结果值，这个结果值就是发给观察者的
        Log.d(TAG, "使用 fromCallable 创建 Observable");
        Observable.fromCallable(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return "Hello World";
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept :: " + s);
            }
        });

        // fromFuture、fromIterable

    }

    @OnClick(R.id.bt7)
    public void mapOperator() {

        // map (将上游的数据转化成下游所需要的任意类型)
        Log.d(TAG, "-------------- 使用 map 转换 --------------");
        Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        emitter.onNext("1");
                    }
                }).map(new Function<String, Integer>() {
            @Override
            public Integer apply(String s) throws Exception {
                return 12345;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept :: " + integer);
            }
        });

        // flatMap
        Log.d(TAG, "-------------- 使用 flatMap 转换 --------------");

        List<Student> students = new ArrayList<>();
        students.add(new Student("Jack", new Course[]{new Course("语文"), new Course("数学")}));
        students.add(new Student("Tom", new Course[]{new Course("英语"), new Course("物理")}));
        students.add(new Student("Kobe", new Course[]{new Course("化学"), new Course("生物")}));

        Observable.fromIterable(students)
                .flatMap(new Function<Student, ObservableSource<Course>>() {
                    @Override
                    public ObservableSource<Course> apply(Student student) throws Exception {
                        Log.d(TAG, "" + student.name);
                        return Observable.fromArray(student.courses);
                    }
                })
                .subscribe(new Consumer<Course>() {
                    @Override
                    public void accept(Course course) throws Exception {
                        Log.d(TAG, "----" + course.name);
                    }
                });

        //
        Log.d(TAG, "-------------- 使用 concatMap 转换 --------------");

        Log.d(TAG, "============= flatMap 延迟效果");
        Observable.fromIterable(students)
                .flatMap(new Function<Student, ObservableSource<Course>>() {
                    @Override
                    public ObservableSource<Course> apply(Student student) throws Exception {
                        Log.d(TAG, "" + student.name);
                        if ("Tom".equals(student.name)) {
                            return Observable.fromArray(student.courses).delay(5, TimeUnit.SECONDS);
                        }
                        return Observable.fromArray(student.courses);
                    }
                })
                .subscribe(new Consumer<Course>() {
                    @Override
                    public void accept(Course course) throws Exception {
                        Log.d(TAG, "----" + course.name);
                    }
                });

        Log.d(TAG, "============= concatMap 延迟效果");
        Observable.fromIterable(students)
                .concatMap(new Function<Student, ObservableSource<Course>>() {
                    @Override
                    public ObservableSource<Course> apply(Student student) throws Exception {
                        Log.d(TAG, "" + student.name);
                        if ("Tom".equals(student.name)) {
                            return Observable.fromArray(student.courses).delay(5, TimeUnit.SECONDS);
                        }
                        return Observable.fromArray(student.courses);
                    }
                })
                .subscribe(new Consumer<Course>() {
                    @Override
                    public void accept(Course course) throws Exception {
                        Log.d(TAG, "----" + course.name);
                    }
                });
    }

    class Student {
        public String name;
        public Course[] courses;

        public Student(String name, Course[] courses) {
            this.name = name;
            this.courses = courses;
        }
    }

    class Course {
        public String name;

        public Course(String name) {
            this.name = name;
        }
    }

    @OnClick(R.id.bt8)
    public void zipOperator() {
        Observable<String> observable1 = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable1 subscribe");
                        emitter.onNext("1");
                        Log.d(TAG, "Observable1 emitter end");
                    }
                }
        );

        Observable<String> observable2 = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable2 subscribe");
                        emitter.onNext("2");
                        Log.d(TAG, "Observable2 emitter end");
                    }
                }
        );

        observable1.zipWith(observable2, new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Exception {
                Log.d(TAG, "s=" + s + " :: " + "s2=" + s2);

                return s + s2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "accept:" + s);
            }
        });


//        Observable.zip(observable1, observable2,
//                new BiFunction<String, String, Integer>() {
//                    @Override
//                    public Integer apply(String s, String s2) throws Exception {
//                        return null;
//                    }
//                })
//                .subscribe(new Consumer<Integer>() {
//                    @Override
//                    public void accept(Integer integer) throws Exception {
//
//                    }
//                });
    }

    @OnClick(R.id.bt9)
    public void doOperator() {
        List<Student> students = new ArrayList<>();
        students.add(new Student("Jack", null));
        students.add(new Student("Tom", null));
        students.add(new Student("Kobe", null));

        Log.d(TAG, "-------------- doOnNext --------------");
        Observable.fromIterable(students)
                .doOnNext(new Consumer<Student>() {
                    @Override
                    public void accept(Student student) throws Exception {
                        Log.d(TAG, "doOnNext accept:" + student.name);

                        student.name = "张三";
                    }
                })
                .subscribe(new Consumer<Student>() {
                    @Override
                    public void accept(Student student) throws Exception {
                        Log.d(TAG, "accept:" + student.name);
                    }
                });

        Log.d(TAG, "-------------- doAfterNext --------------");
        Observable.fromIterable(students)
                .doAfterNext(new Consumer<Student>() {
                    @Override
                    public void accept(Student student) throws Exception {
                        Log.d(TAG, "doAfterNext accept:" + student.name);
                    }
                })
                .doOnNext(new Consumer<Student>() {
                    @Override
                    public void accept(Student student) throws Exception {
                        Log.d(TAG, "doOnNext accept:" + student.name);

                        student.name = "张三";
                    }
                })
                .subscribe(new Consumer<Student>() {
                    @Override
                    public void accept(Student student) throws Exception {
                        Log.d(TAG, "accept:" + student.name);
                    }
                });

        Log.d(TAG, "-------------- doOnSubscribe doOnComplete doOnError doFinally --------------");
        // 作用：组合多个被观察者一起发送数据，合并后 按发送顺序串行执行  注意：concat（）组合被观察者数量≤4个

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.d(TAG, "subscribe ... ");
                emitter.onNext("Hello World");
                emitter.onComplete();
            }
        })
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.d(TAG, "doOnSubscribe ... ");
                    }
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.d(TAG, "doOnComplete ... ");
                    }
                })
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG, "doOnError ... ");
                    }
                })
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.d(TAG, "doFinally ... ");
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(TAG, "accept:" + s);
                    }
                });
    }

    @OnClick(R.id.bt10)
    public void concatOperator() {
        Observable<String> observable1 = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable1 subscribe");
                        emitter.onNext("1");
                        emitter.onComplete();
                        Log.d(TAG, "Observable1 emitter end");
                    }
                }
        );

        Observable<String> observable2 = Observable.create(
                new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.d(TAG, "Observable2 subscribe");
                        emitter.onNext("2");
                        emitter.onComplete();
                        Log.d(TAG, "Observable2 emitter end");
                    }
                }
        );

        Observable<Integer> observable3 = Observable.create(
                new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        Log.d(TAG, "Observable3 subscribe");
                        emitter.onNext(88);
                        emitter.onComplete();
                        Log.d(TAG, "Observable3 emitter end");
                    }
                }
        );

        Log.d(TAG, "-------------- concat --------------");

        Observable.concat(observable1, observable2)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(TAG, "accept:" + s);
                    }
                });

        Observable.concat(Observable.just(1)
                , Observable.just(2)
                , Observable.just(3)
                , Observable.just(4))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });

        Observable.concat(observable1, observable3)
                .subscribe(new Consumer<Serializable>() {
                    @Override
                    public void accept(Serializable serializable) throws Exception {

                    }
                });

        Log.d(TAG, "-------------- concatArray --------------");
        // 组合多个被观察者一起发送数据，合并后 按发送顺序串行执行  注意：concatArray（）则可＞4个

        Observable.concatArray(
                Observable.just(1)
                , Observable.just(2)
                , Observable.just(3)
                , Observable.just(4)
                , Observable.just(5)
                , Observable.just(6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });

        Log.d(TAG, "-------------- merge --------------");
        // 组合多个被观察者一起发送数据，合并后 按时间线并行执行,组合被观察者的数量，即merge（）组合被观察者数量≤4个
        Observable.merge(
                Observable.intervalRange(1, 3, 2, 1, TimeUnit.SECONDS),
                Observable.intervalRange(4, 3, 2, 1, TimeUnit.SECONDS))
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d(TAG, "accept:" + aLong);
                    }
                });

        // 组合被观察者的数量，即merge（）组合被观察者数量≤4个，而mergeArray（）则可＞4个

        Log.d(TAG, "-------------- concatArrayDelayError --------------");

        Observable.concat(
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onNext(3);
                        emitter.onError(new Exception());
                    }
                }), Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG, "Error ...");
                    }
                });

        // onError事件推迟到其他观察者发送事件结束
        Observable.concatArrayDelayError(
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        emitter.onNext(1);
                        emitter.onNext(2);
                        emitter.onNext(3);
                        emitter.onError(new Exception());
                    }
                }), Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG, "Error ...");
                    }
                });

    }

    int i = 10;

    @OnClick(R.id.bt11)
    public void deferUser() {
        // 这个方法的作用就是直到被观察者被订阅后才会创建被观察者
        Observable<Integer> observable =
                Observable.defer(
                        new Callable<ObservableSource<? extends Integer>>() {
                            @Override
                            public ObservableSource<? extends Integer> call() throws Exception {
                                return Observable.just(i);
                            }
                        });

        i = 100;

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "accept:" + integer);
            }
        };

        observable.subscribe(consumer);

        i = 88;

        observable.subscribe(consumer);
    }

    @OnClick(R.id.bt12)
    public void timerUser() {
        // 当到指定时间后就会发送一个 0L 的值给观察者
        Observable
                .timer(3, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.d(TAG, "accept:" + aLong);
                    }
                });
    }

    @OnClick(R.id.bt13)
    public void intervalUser() {
        // 每隔一段时间就会发送一个事件，这个事件是从0开始，不断增1的数字
//        Observable
//                .interval(3, TimeUnit.SECONDS)
//                .subscribe(new Consumer<Long>() {
//                    @Override
//                    public void accept(Long aLong) throws Exception {
//                        Log.d(TAG, "accept:" + aLong);
//                    }
//                });

        // initialDelay 参数，这个参数的意思就是 onSubscribe 回调之后，再次回调 onNext 的间隔时间
//        Observable
//                .interval(5, 2, TimeUnit.SECONDS)
//                .subscribe(new Consumer<Long>() {
//                    @Override
//                    public void accept(Long aLong) throws Exception {
//                        Log.d(TAG, "accept:" + aLong);
//                    }
//                });

        // 可以指定发送事件的开始值和数量，其他与 interval() 的功能一样。
//        Observable
//                .intervalRange(1, 5, 0, 3, TimeUnit.SECONDS)
//                .subscribe(new Consumer<Long>() {
//                    @Override
//                    public void accept(Long aLong) throws Exception {
//                        Log.d(TAG, "accept:" + aLong);
//                    }
//                });

        // 同时发送一定范围的事件序列
        Observable
                .range(2, 8)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
        // rangeLong 作用与 range() 一样，只是数据类型为 Long

    }

    @OnClick(R.id.bt14)
    public void empty_never_error_User() {
//        empty() ： 直接发送 onComplete() 事件
//        never()：不发送任何事件
//        error()：发送 onError() 事件

        Observer<Integer> observer =
                new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "accept:" + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                };

        Log.d(TAG, "-------------- empty --------------");
        Observable
                .<Integer>empty()
                .subscribe(observer);

        Log.d(TAG, "-------------- never --------------");
        Observable
                .<Integer>never()
                .subscribe(observer);

        Log.d(TAG, "-------------- error --------------");
        Observable
                .<Integer>error(new Exception("Runtime exception"))
                .subscribe(observer);
    }

    @OnClick(R.id.bt14)
    public void bufferUser() {
        // 从需要发送的事件当中获取一定数量的事件，并将这些事件放到缓冲区当中一并发出
        Observable.just(1, 3, 5, 7, 9, 11)
                .buffer(2, 1)
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Exception {
                        Log.d(TAG, "================缓冲区大小： " + integers.size());
                        for (Integer i : integers) {
                            Log.d(TAG, "================元素： " + i);
                        }
                    }
                });
    }

    @OnClick(R.id.bt16)
    public void groupByUse() {
        // 将发送的数据进行分组，每个分组都会返回一个被观察者
        Observable
                .just(10, 9, 8, 7, 6, 1, 2, 3, 4, 5)
                .groupBy(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        return integer % 2;
                    }
                })
                .subscribe(new Consumer<GroupedObservable<Integer, Integer>>() {
                    @Override
                    public void accept(final GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) throws Exception {
                        integerIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.d(TAG, "key:" + integerIntegerGroupedObservable.getKey() + " == value:" + integer);
                            }
                        });
                    }
                });
    }

    @OnClick(R.id.bt17)
    public void scanUse() {
        // 将数据以一定的逻辑聚合起来
        Observable
                .just(1, 2, 3, 4, 5)
                .scan(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        Log.d(TAG, "integer:" + integer + "  integer2:" + integer2);
                        return integer + integer2;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    @OnClick(R.id.bt18)
    public void windowUse() {
        // 发送指定数量的事件时，就将这些事件分为一组。window 中的 count 的参数就是代表指定的数量，
        // 例如将 count 指定为2，那么每发2个数据就会将这2个数据分成一组。

        Observable
                .just(1, 2, 3, 4, 5)
                .window(2)
                .subscribe(new Consumer<Observable<Integer>>() {
                    @Override
                    public void accept(final Observable<Integer> integerObservable) throws Exception {
                        integerObservable
                                .doOnComplete(new Action() {
                                    @Override
                                    public void run() throws Exception {
                                        Log.d(TAG, "doOnComplete ... ");
                                    }
                                })
                                .subscribe(new Consumer<Integer>() {
                                    @Override
                                    public void accept(Integer integer) throws Exception {
                                        Log.d(TAG, "accept:" + integer);
                                    }
                                });
                    }
                });
    }

    @OnClick(R.id.bt19)
    public void reduceUse() {
        // 与 scan() 操作符的作用也是将发送数据以一定逻辑聚合起来，这两个的区别在于 scan() 每处理一次数据就会将事件发送给观察者，
        // 而 reduce() 会将所有数据聚合在一起才会发送事件给观察者。
        Observable
                .just(1, 2, 3, 4, 5)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        Log.d(TAG, "integer:" + integer + "  integer2:" + integer2);
                        return integer + integer2;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    @OnClick(R.id.bt20)
    public void collectUse() {
        // 将数据收集到数据结构当中。
        Observable
                .just(1, 2, 3, 4, 5)
                .collect(new Callable<ArrayList<Integer>>() {
                    @Override
                    public ArrayList<Integer> call() throws Exception {
                        return new ArrayList<>();
                    }
                }, new BiConsumer<ArrayList<Integer>, Integer>() {
                    @Override
                    public void accept(ArrayList<Integer> integers, Integer integer) throws Exception {
                        integers.add(integer);
                    }
                })
                .subscribe(new Consumer<ArrayList<Integer>>() {
                    @Override
                    public void accept(ArrayList<Integer> integers) throws Exception {
                        Log.d(TAG, "accept:" + integers);
                    }
                });
    }

    @OnClick(R.id.bt21)
    public void startWithUse() {
        // 在发送事件之前追加事件，startWith() 追加一个事件，startWithArray() 可以追加多个事件。追加的事件会先发出。
        Observable
                .just(1, 2, 3, 4, 5)
                .startWithArray(6, 7, 8, 9)
                .startWith(10)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    /***************************************************************************/

    /*
     * 通过一定逻辑来过滤被观察者发送的事件，如果返回 true 则会发送事件，否则不会发送。
     * */
    @OnClick(R.id.bt501)
    public void filterOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer > 1 && integer < 5;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 可以过滤不符合该类型事件
     * */
    @OnClick(R.id.bt502)
    public void ofTypeOperator() {
        Observable
                .just(1, 2, 3, "China", "America")
                .ofType(String.class)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(TAG, "accept:" + s);
                    }
                });
    }

    /*
     * 跳过正序某些事件，count 代表跳过事件的数量
     * */
    @OnClick(R.id.bt503)
    public void skipOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .skip(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 过滤事件序列中的重复事件。
     * */
    @OnClick(R.id.bt504)
    public void distinctOperator() {
        Observable
                .just(1, 2, 2, 3, 5, 5, 9, 3)
                .distinct()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 过滤掉连续重复的事件
     * */
    @OnClick(R.id.bt505)
    public void distinctUntilChangedOperator() {
        Observable
                .just(1, 2, 2, 5, 3, 5, 2, 9)
                .distinctUntilChanged()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 控制观察者接收的事件的数量
     * */
    @OnClick(R.id.bt506)
    public void takeOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .take(3)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 如果两件事件发送的时间间隔小于设定的时间间隔则前一件事件就不会发送给观察者。
     * */
    @OnClick(R.id.bt507)
    public void debounceOperator() {
        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) {
                        try {
                            emitter.onNext(1);
                            Thread.sleep(2000);
                            emitter.onNext(2);
                            Thread.sleep(3100);
                            emitter.onNext(3);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                })
                .debounce(3, TimeUnit.SECONDS)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    /*
     * 取事件序列的第一个元素
     * */
    @OnClick(R.id.bt508)
    public void firstElementOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .firstElement()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 取事件序列的最后一个元素
     * */
    @OnClick(R.id.bt509)
    public void lastElementOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .lastElement()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    /*
     * 可以指定取出事件序列中事件，但是输入的 index 超出事件序列的总数的话就不会出现任何结果。
     * 这种情况下，你想发出异常信息的话就用 elementAtOrError() 。
     * */
    @OnClick(R.id.bt510)
    public void elementAtOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .elementAt(3)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    /***************************************************************************/


    /*
     * 判断事件序列是否全部满足某个事件，如果都满足则返回 true，反之则返回 false。
     * */
    @OnClick(R.id.bt601)
    public void allOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .all(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer < 8;
                    }
                })
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean aBoolean) throws Exception {
                        Log.d(TAG, "accept:" + aBoolean);
                    }
                });
    }

    /*
     * 可以设置条件，当某个数据满足条件时就会发送该数据，反之则不发送。
     * */
    @OnClick(R.id.bt602)
    public void takeWhileOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .takeWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer > 0;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    /*
     * 可以设置条件，当某个数据满足条件时不发送该数据，反之则发送。
     * */
    @OnClick(R.id.bt603)
    public void skipWhileOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .skipWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer > 0;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 可以设置条件，当事件满足此条件时，下一次的事件就不会被发送了。
     * */
    @OnClick(R.id.bt604)
    public void takeUntilOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .takeUntil(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer > 3;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 当 skipUntil() 中的 Observable 发送事件了，原来的 Observable 才会发送事件给观察者
     * */
    @OnClick(R.id.bt605)
    public void skipUntilOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .skipUntil(Observable.intervalRange(6, 5, 3, 1, TimeUnit.SECONDS))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 判断两个 Observable 发送的事件是否相同。
     * */
    @OnClick(R.id.bt606)
    public void sequenceEqualOperator() {
        Observable
                .sequenceEqual(
                        Observable.just(1, 2, 3, 4, 5),
                        Observable.just("China", "America"))
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean aBoolean) throws Exception {
                        Log.d(TAG, "accept:" + aBoolean);
                    }
                });
    }

    /*
     * 判断事件序列中是否含有某个元素，如果有则返回 true，如果没有则返回 false。
     * */
    @OnClick(R.id.bt607)
    public void containsOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .contains(3)
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean aBoolean) throws Exception {
                        Log.d(TAG, "accept:" + aBoolean);
                    }
                });
    }

    /*
     * 判断事件序列是否为空
     * */
    @OnClick(R.id.bt608)
    public void isEmptyOperator() {
        Observable
                .just(1, 2, 3, 4, 5)
                .isEmpty()
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean aBoolean) throws Exception {
                        Log.d(TAG, "accept:" + aBoolean);
                    }
                });
    }


    /*
     * amb() 要传入一个 Observable 集合，但是只会发送最先发送事件的 Observable 中的事件，其余 Observable 将会被丢弃。
     * */
    @OnClick(R.id.bt609)
    public void ambOperator() {
        Observable<Integer> observable1 = Observable.just(1, 2, 3);
        Observable<Integer> observable2 = Observable.just(4, 5, 6);
        List<Observable<Integer>> observables = new ArrayList<>();
        observables.add(observable1);
        observables.add(observable2);

        Observable.amb(observables)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }

    /*
     * 如果观察者只发送一个 onComplete() 事件，则可以利用这个方法发送一个值。
     * */
    @OnClick(R.id.bt610)
    public void defaultIfEmptyOperator() {
        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) {
                        emitter.onComplete();
                    }
                })
                .defaultIfEmpty(0)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    /***************************************************************************/

    @OnClick(R.id.bt701)
    public void notUseFlowable() { // 不使用背压，内存泄漏
        Observable
                .create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                        int i = 0;
                        while (true) {
                            i++;
                            e.onNext(i);
                        }
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Thread.sleep(5000);
                        Log.d(TAG, "accept:" + integer);
                    }
                });
    }


    @OnClick(R.id.bt702)
    public void useFlowable() {
        Flowable
                .create(new FlowableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                        int i = 0;
                        while (true) {
                            if (emitter.requested() == 0) continue; // 请求量为 0 时，不发送数据
                            i++;
                            emitter.onNext(i);
                        }
                    }
                }, BackpressureStrategy.MISSING)
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.newThread())
                .subscribe(new Subscriber<Integer>() {

                    Subscription mSubscription;

                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription = s;
                        mSubscription.request(1); // 初始发送量设置为 1
                    }

                    @Override
                    public void onNext(Integer integer) {
                        try {
                            Thread.sleep(500);
                            Log.d(TAG, "accept:" + integer);
                            mSubscription.request(1); // 每接收一条数据，增加一个发送量
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onError(Throwable t) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }


}
